
package com.hery.core;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;

import java.net.InetSocketAddress;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.hery.common.Constants;
import com.hery.common.NamedThreadFactory;
import com.hery.common.Version;



public class NettyClient extends AbstractEndpoint {
    
    private static final Logger logger = LoggerFactory.getLogger(NettyClient.class);


    private Bootstrap bootstrap;
    private EventLoopGroup group=new NioEventLoopGroup();

    private volatile io.netty.channel.Channel channel; // volatile, please copy reference to use
    protected static final String CLIENT_THREAD_POOL_NAME  ="HeryClientHandler";
    
    private static final AtomicInteger CLIENT_THREAD_POOL_ID = new AtomicInteger();

    private final Lock            connectLock = new ReentrantLock();
    
    private static final ScheduledThreadPoolExecutor reconnectExecutorService = new ScheduledThreadPoolExecutor(2, new NamedThreadFactory("DubboClientReconnectTimer", true));
    
    private volatile  ScheduledFuture<?> reconnectExecutorFuture = null;

    
    private final AtomicInteger reconnect_count = new AtomicInteger(0);
    
    //重连的error日志是否已经被调用过.
    private final AtomicBoolean reconnect_error_log_flag = new AtomicBoolean(false) ;

    public NettyClient(String host,int port) throws RemotingException{
    	
        setHost(host);
        setPort(port);
        try {
            doOpen();
        } catch (Throwable t) {
            close();
            throw new RemotingException("Failed to start , connect to the server " + getRemoteAddress() + ", cause: " + t.getMessage(), t);
        }
        try {
            // connect.
            connect();
            if (logger.isInfoEnabled()) {
                logger.info("Start , connect to the server " + getRemoteAddress());
            }
        } catch (Throwable t){
            close();
            throw new RemotingException("Failed to start ,connect to the server " + getRemoteAddress() + ", cause: " + t.getMessage(), t);
        }
    }
    
    protected void doOpen() throws Throwable {
        bootstrap = new Bootstrap();
        bootstrap.group(group);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
        bootstrap.option(ChannelOption.TCP_NODELAY, true);
        bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getTimeout());
        final NettyHandler nettyHandler = new NettyHandler();
        bootstrap.handler(new ChannelInitializer(){
			@Override
			protected void initChannel(io.netty.channel.Channel nChannel) throws Exception {
				ChannelPipeline pipeline= nChannel.pipeline();
				pipeline.addLast(new LengthFieldBasedFrameDecoder(100000000,0,4,0,4));
				pipeline.addLast(nettyHandler);
			}
        	
        });
    }

    protected void doConnect() throws Throwable {
        long start = System.currentTimeMillis();
        ChannelFuture future = bootstrap.connect(getConnectAddress());
        try{
            boolean ret = future.awaitUninterruptibly(getConnectTimeout(), TimeUnit.MILLISECONDS);
            
            if (ret && future.isSuccess()) {
                io.netty.channel.Channel newChannel = future.channel();

                try {
                    // 关闭旧的连接
                	io.netty.channel.Channel oldChannel = NettyClient.this.channel; 
                    if (oldChannel != null) {
                        try {
                            if (logger.isInfoEnabled()) {
                                logger.info("Close old netty channel " + oldChannel + " on create new netty channel " + newChannel);
                            }
                            oldChannel.close();
                        } finally {
                            NettyChannel.removeChannelIfDisconnected(oldChannel);
                        }
                    }
                } finally {
                    if (this.isClosed()) {
                        try {
                            if (logger.isInfoEnabled()) {
                                logger.info("Close new netty channel " + newChannel + ", because the client closed.");
                            }
                            newChannel.close();
                        } finally {
                            NettyClient.this.channel = null;
                            NettyChannel.removeChannelIfDisconnected(newChannel);
                        }
                    } else {
                        NettyClient.this.channel = newChannel;
                    }
                }
            } else if (future.cause() != null) {
                throw new RemotingException("client failed to connect to server "
                        + getRemoteAddress() + ", error message is:" + future.cause().getMessage(), future.cause());
            } else {
                throw new RemotingException("failed to connect to server "
                        + getRemoteAddress() + " client-side timeout "
                        + getConnectTimeout() + "ms (elapsed: " + (System.currentTimeMillis() - start) + "ms)  using hery version " + Version.getVersion());
            }
        }finally{
            if (! isConnected()) {
                future.cancel(true);
            }
        }
    }

    protected void doDisConnect() throws Throwable {
        try {
            NettyChannel.removeChannelIfDisconnected(channel);
        } catch (Throwable t) {
            logger.warn(t.getMessage());
        }
    }
    
    protected void doClose() throws Throwable {
        /*try {
            bootstrap.releaseExternalResources();
        } catch (Throwable t) {
            logger.warn(t.getMessage());
        }*/
    }

    protected NettyChannel getChannel() {
        io.netty.channel.Channel c = channel;
        if (c == null || ! c.isActive())
            return null;
        return NettyChannel.getOrAddChannel(c);
    }
   



    
    private synchronized void destroyConnectStatusCheckCommand(){
        try {
            if (reconnectExecutorFuture != null && ! reconnectExecutorFuture.isDone()){
                reconnectExecutorFuture.cancel(true);
                reconnectExecutorService.purge();
            }
        } catch (Throwable e) {
            logger.warn(e.getMessage(), e);
        }
    }
 
    
    public InetSocketAddress getConnectAddress() {
        return new InetSocketAddress(getHost(), getPort());
    }

    public InetSocketAddress getRemoteAddress() {
    	NettyChannel channel = getChannel();
        if (channel == null)
            return new InetSocketAddress(getHost(), getPort());
        return channel.getRemoteAddress();
    }



    public boolean isConnected() {
    	NettyChannel channel = getChannel();
        if (channel == null)
            return false;
        return channel.isConnected();
    }

    public Object getAttribute(String key) {
    	NettyChannel channel = getChannel();
        if (channel == null)
            return null;
        return channel.getAttribute(key);
    }

    public void setAttribute(String key, Object value) {
    	NettyChannel channel = getChannel();
        if (channel == null)
            return;
        channel.setAttribute(key, value);
    }

    public void removeAttribute(String key) {
    	NettyChannel channel = getChannel();
        if (channel == null)
            return;
        channel.removeAttribute(key);
    }

    public boolean hasAttribute(String key) {
    	NettyChannel channel = getChannel();
        if (channel == null)
            return false;
        return channel.hasAttribute(key);
    }
    
    public void send(Object message, boolean sent) throws RemotingException {
        if ( !isConnected()){
            connect();
        }
        NettyChannel channel = getChannel();
        //TODO getChannel返回的状态是否包含null需要改进
        if (channel == null || ! channel.isConnected()) {
          throw new RemotingException( "message can not send, because channel is closed . :" + getRemoteAddress());
        }
        channel.send(message, sent);
    }

    protected void connect() throws RemotingException {
        connectLock.lock();
        try {
            if (isConnected()) {
                return;
            }
            doConnect();
            if (! isConnected()) {
                throw new RemotingException("Failed connect to server " + getRemoteAddress() + "  using hery version " + Version.getVersion()
                                            + ", cause: Connect wait timeout: " + getTimeout() + "ms.");
            } else {
            	if (logger.isInfoEnabled()){
            		logger.info("Successed connect to server " + getRemoteAddress() + "  using hery version " + Version.getVersion()
                                            + ", channel is " + this.getChannel());
            	}
            }
            reconnect_count.set(0);
            reconnect_error_log_flag.set(false);
        } catch (RemotingException e) {
            throw e;
        } catch (Throwable e) {
            throw new RemotingException("Failed connect to server " + getRemoteAddress() + "  using hery version " + Version.getVersion()
                                        + ", cause: " + e.getMessage(), e);
        } finally {
            connectLock.unlock();
        }
    }

    public void disconnect() {
        connectLock.lock();
        try {
            destroyConnectStatusCheckCommand();
            try {
            	NettyChannel channel = getChannel();
                if (channel != null) {
                    channel.close();
                }
            } catch (Throwable e) {
                logger.warn(e.getMessage(), e);
            }
            try {
                doDisConnect();
            } catch (Throwable e) {
                logger.warn(e.getMessage(), e);
            }
        } finally {
            connectLock.unlock();
        }
    }
    
    public void reconnect() throws RemotingException {
        disconnect();
        connect();
    }

    public void close() {

    	try {
            super.close();
        } catch (Throwable e) {
            logger.warn(e.getMessage(), e);
        }
        try {
        	disconnect();
        } catch (Throwable e) {
            logger.warn(e.getMessage(), e);
        }
        try {
            doClose();
        } catch (Throwable e) {
            logger.warn(e.getMessage(), e);
        }
    }

  

}